package com.ztesoft.web.checkup.service.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import javax.sql.DataSource;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;

import com.ztesoft.core.db.ds.DynamicDataSourceManager;
import com.ztesoft.framework.exception.BaseAppException;
import com.ztesoft.framework.log.ZTEsoftLogManager;
import com.ztesoft.framework.util.DateUtils;
import com.ztesoft.framework.util.Utils;
import com.ztesoft.web.baseconfig.db.po.DmPlanPO;
import com.ztesoft.web.baseconfig.db.po.DmSrcTablePO;
import com.ztesoft.web.baseconfig.db.po.DmStrategyPO;
import com.ztesoft.web.checkup.CheckUpTask;
import com.ztesoft.web.checkup.service.CheckUpStrategy;
import com.ztesoft.web.common.DMSConstant;
import com.ztesoft.web.common.Util;
import com.ztesoft.web.common.customdb.SQLHelper;
import com.ztesoft.web.common.customdb.dao.DmTaskStageCustomDao;
import com.ztesoft.web.common.customdb.dao.dto.DmTaskStageDto;
import com.ztesoft.web.resourcecheck.ResourceCheckJob;
import com.ztesoft.web.syncbaseinfo.service.BaseInfoService;
import com.ztesoft.web.taskgenerator.db.dao.DmTaskDao;
import com.ztesoft.web.taskgenerator.db.po.DmTaskPO;


/**
 * 体检策略<br>
 * 按条件执行体检
 * 
 * @author <a href="mailto:lin.weibin@zte.com.cn">lin.wb</a> 2016年5月19日 下午7:13:20
 * @version 1.0
 */


public class CheckUpStrategyCondition implements CheckUpStrategy {
	
	private static final ZTEsoftLogManager LOGGER = ZTEsoftLogManager
	            .getLogger(CheckUpStrategyCondition.class);
	
	@Autowired
	private DynamicDataSourceManager dynamicDataSourceManager;
	
	@Autowired
	private BaseInfoService baseInfoService;
	
	@Autowired
	private DmTaskStageCustomDao taskStageCustomDao;

	@Autowired
	private ResourceCheckJob resourceCheckJob;
	
	@Autowired
	private DmTaskDao taskDao;
	
	public String checkUp(DmPlanPO planPO, DmTaskPO taskPO, DmStrategyPO strategyPO, DmSrcTablePO srcTablePO) {
		String checkState = DMSConstant.Status.SUCCESS.getState();
		LOGGER.debug("|查询任务环节列表|strategyId={0},taskId={1}", strategyPO.getDmStrategyId(), taskPO.getDmTaskId());
		List<DmTaskStageDto> taskStageList = taskStageCustomDao.selectTaskStageList(taskPO.getDmTaskId(), 
				strategyPO.getDmStrategyId(), DMSConstant.STAGE_CHECKUP,"00T");
		LOGGER.info("|查询待体检任务环节列表|\r\n{0}",taskStageList);
		if (Utils.isEmpty(taskStageList)) {
			return checkState;//
		}
		String[] srcDbConfig = null;
		DataSource srcDataSource = null;
		try {
			srcDbConfig = baseInfoService.getDbConfig(srcTablePO.getDbLinkId(), srcTablePO.getDbUserName());
			srcDataSource = dynamicDataSourceManager.getDruidDataSource(srcTablePO.getDbLinkId()+srcTablePO.getDbUserName(), srcDbConfig);
		} catch (BaseAppException e) {
			checkState = DMSConstant.Status.FALSE.getState();
			LOGGER.error("",e);
			DmTaskStageDto po = taskStageList.get(0);
			po.setTaskStartTime(DateUtils.getDBCurrentTime());
			po.setState(checkState);
			po.setStateDate(DateUtils.getDBCurrentTime());
			po.setTaskEndTime(DateUtils.getDBCurrentTime());
			po.setCheckObjName(null);//更新策略下所有
			po.setStage(null);
			po.setErrorDesc(e.getLocaleMessage());
			taskStageCustomDao.updateTaskStageState(po);//
			return checkState;
		}
		
		List<CheckUpTask> checkUpTaskList = new ArrayList<CheckUpTask>();
		for (DmTaskStageDto taskStageDto : taskStageList) {
			String sqlCountStr = SQLHelper.selectCount(taskStageDto.getCheckObjName(), srcTablePO.getSelectCondition(), //统计sql处理，只体检不范围统计~modified by lwb 2015/12/29
					DMSConstant.TABLE_CHECKUP != (int)planPO.getOperType() ? String.valueOf(strategyPO.getUpperCnt()) : null, srcDbConfig[0]);
			LOGGER.info("|统计{0}SQL语句{1},dm_task_stage_id={2}|", taskStageDto.getCheckObjName(), sqlCountStr, taskStageDto.getDmInfoId());
			String sqlStr = SQLHelper.selectPageByCondition(srcTablePO.getKeyColumn(), taskStageDto.getCheckObjName(), srcTablePO.getSelectCondition(), srcDbConfig[0]);
			LOGGER.info("|分页查询{0}SQL语句{1},,dm_task_stage_id={2}", taskStageDto.getCheckObjName(), sqlStr, taskStageDto.getDmInfoId());
			
			try {
				int rowsCount = SQLHelper.countRows(new JdbcTemplate(srcDataSource), sqlCountStr);
				if (rowsCount > (int)strategyPO.getUpperCnt()&& DMSConstant.TABLE_CHECKUP != (int)planPO.getOperType()){
					rowsCount = strategyPO.getUpperCnt();
				}
				taskStageDto.setTotalRecordCount(rowsCount);
				taskStageDto.setTotalDataSize(Util.divValue2BigDecimal(srcTablePO.getOneRecordSize(), rowsCount, 2));
				taskStageCustomDao.updateTaskStage(taskStageDto);//
			} catch (Exception e) {
				checkState = DMSConstant.Status.FALSE.getState();
				LOGGER.error("|体检异常|"+taskStageDto.getCheckObjName()+"表END|",e);
				taskStageDto.setTaskStartTime(DateUtils.getDBCurrentTime());//插入当前时间
				taskStageDto.setTaskEndTime(DateUtils.getDBCurrentTime());
				taskStageDto.setState(checkState);
				taskStageDto.setStage(null);
				taskStageDto.setErrorDesc(e.getMessage());
				taskStageCustomDao.updateTaskStageState(taskStageDto);//
				LOGGER.error("|体检异常|托送告警信息-----");
				resourceCheckJob.pushMsg2IOCP("[checkup table{"+taskStageDto.getCheckObjName()+"} error]strategyID:"+taskStageDto.getDmStrategyId(), taskStageDto.getErrorDesc(), "体检失败");
				continue;
			}
			
			LOGGER.info("|待体检{0}数据量|TotalRecordCount={1}", taskStageDto.getCheckObjName(), taskStageDto.getTotalRecordCount());
			if (taskStageDto.getTotalRecordCount() == 0) {
				checkState = DMSConstant.Status.DATA_NULL.getState();
				taskStageDto.setTaskStartTime(DateUtils.getDBCurrentTime());
				taskStageDto.setState(DMSConstant.Status.SUCCESS.getState());
				taskStageDto.setStateDate(DateUtils.getDBCurrentTime());
				taskStageDto.setTaskEndTime(DateUtils.getDBCurrentTime());
				taskStageDto.setStage(null);
				taskStageDto.setErrorDesc(null);
				taskStageDto.setIsCheck(DMSConstant.CHECK_PASSED);//默认审批通过
				taskStageCustomDao.updateTaskStageState(taskStageDto);//
				continue; //return "50E"; 20160225 modified by lin.wb
			}
			if ((int)taskStageDto.getTotalRecordCount() == (int)strategyPO.getUpperCnt()) {
				taskPO.setNeedSubtask(DMSConstant.NEED_SUBTASK_T);
				taskPO.setStateDate(DateUtils.getDBCurrentTime());
				taskDao.updateByPrimaryKey(taskPO);//更新是否需要子任务
			}
			checkUpTaskList.add(new CheckUpTask(sqlStr, srcTablePO, taskStageDto, taskPO.getDmTaskId(), strategyPO, planPO, srcDbConfig));
			
		}
		
		// 体检数据为空 或 统计数据异常
		if ((DMSConstant.Status.DATA_NULL.getState().equals(checkState)
				||DMSConstant.Status.FALSE.getState().equals(checkState)) && Utils.isEmpty(checkUpTaskList)){
			return checkState;
		}
		
		//ExecutorService线程处理
		int concurrentCount = strategyPO.getConcurrentNumber()>=taskStageList.size() ? 
				taskStageList.size() : strategyPO.getConcurrentNumber();
		return execute(strategyPO, srcTablePO, checkState, concurrentCount, checkUpTaskList);
		
	}

	private String execute(DmStrategyPO strategyPO, DmSrcTablePO srcTablePO,String checkState, int nThreads,
			List<CheckUpTask> checkUpTaskList) {
		List<Future<DmTaskStageDto>> futures = null;
		ExecutorService executor = null;
		int sucessInt = 0,failInt = 0,stopInt = 0, sysStopInt = 0, cancelInt = 0;
		try {
			executor = Executors.newFixedThreadPool(nThreads);
			futures = executor.invokeAll(checkUpTaskList);
			for (Future<DmTaskStageDto> future : futures) {
				checkState = future.get().getCheckState();
				LOGGER.info("|体检|{0}表END|:::{1}",future.get().getCheckObjName(), checkState);
				if (DMSConstant.SUCCESS.equals(checkState)) {
					future.get().setState(DMSConstant.Status.SUCCESS.getState());
					sucessInt = sucessInt + 1;
				} else if (DMSConstant.FAIL.equals(checkState)){
					future.get().setState(DMSConstant.Status.FALSE.getState());
					future.get().setStage(null);
					failInt = failInt + 1;
				} else if (DMSConstant.Status.CANCEL.getState().equals(checkState)) {
					cancelInt = cancelInt + 1;
				} else if (DMSConstant.Status.PAUSE.getState().equals(checkState)) {
					stopInt = stopInt + 1;
				} else if (DMSConstant.Status.SYSPAUSE.getState().equals(checkState)) {
					sysStopInt = sysStopInt + 1;
				}
				
				if (DMSConstant.SUCCESS.equals(checkState)|| DMSConstant.FAIL.equals(checkState)) {//更新环节
					future.get().setTaskStartTime(null);//避免数据插入
					future.get().setTaskEndTime(DateUtils.getDBCurrentTime());
					future.get().setStateDate(DateUtils.getDBCurrentTime());
					future.get().setCheckDataSize(Util.divValue2BigDecimal(srcTablePO.getOneRecordSize(), future.get().getTotalRecordCount(), 2));
					taskStageCustomDao.updateTaskStageState(future.get());
				}
			}
			LOGGER.debug("checkup{0} end...",strategyPO.getDmStrategyId());
			LOGGER.info("|++===体检服务|任务END..sucessInt:{0},failInt:{1},stopInt:{2},sysStopInt:{3},cancelInt:{4}",sucessInt,failInt,stopInt,sysStopInt,cancelInt);
			if (cancelInt > 0){
				checkState = DMSConstant.Status.CANCEL.getState();//取消
			} else if (sysStopInt > 0) {
				checkState = DMSConstant.Status.SYSPAUSE.getState();//系统暂停
			} else if (stopInt > 0) {
				checkState = DMSConstant.Status.PAUSE.getState();//暂停
			} else if (failInt > 0) {
				checkState = DMSConstant.Status.FALSE.getState();//失败
			} else {
				checkState = DMSConstant.Status.SUCCESS.getState();//成功
			}
		} catch (InterruptedException e) {
			LOGGER.error("", e);
			checkState = DMSConstant.Status.FALSE.getState();
		} catch (ExecutionException e) {
			LOGGER.error("", e);
			checkState = DMSConstant.Status.FALSE.getState();
		} finally {
			executor.shutdown();
		}
		return checkState;
	};
	
}
